package org.rabbitmq.mq02.mqServer;

import org.rabbitmq.mq02.common.*;
import org.rabbitmq.mq02.mqServer.core.BasicProperties;
import org.rabbitmq.mq02.mqServer.core.Binding;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BrokerServer {
    //调用相关数据
    private VirtualHost virtualHost = new VirtualHost("default");
    //服务器⾃⾝的 socket
    private ServerSocket serverSocket = null;
    //引入线程池,处理多个客户端的请求
    private ExecutorService executorService = null;
    //引入一个哈希表,存储所有的会话对象
    //key: channelId, val:socket对象
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();
    //引入一个布尔变量,表示当前服务器是否要停止,
    //要对所有线程是立即可见的,用volatile修饰
    private volatile boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(9090);
    }

    //启动服务
    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动服务");
        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                //accept:不断接收客户端发来的请求:
                Socket clientSocket = serverSocket.accept();
                executorService.submit(() -> {
                    processConnection(clientSocket);
                });
            }
        } catch (SocketException e) {
            //正常结束
            System.out.println("[BrokerServer] 服务器停止运行!");
        }
    }

    //停止服务器
    public void stop() throws IOException {
        runnable = false;
        executorService.shutdown();
        serverSocket.close();
    }

    //处理一个客户端的连接
    //一个个连接可能有多次的请求和相应
    //要读取数据,处理数据,然后将结果返回给客户端
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            //  这里需要按照特定格式进行读取和解析数据
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while (true) {
                    //1.读取请求
                    Request request = readRequest(dataInputStream);
                    //2.根据请求计算相应
                    Response response = process(request, clientSocket);
                    //3.将结果返回给客户端
                    writeResponse(dataOutputStream,response);
                }
            } catch (EOFException | SocketException e) {
                //当出现这两种异常时,是正常的异常,是请求读取结束了,读到了空字符串抛出的异常,
                // 正常结束循环就可以了
                System.out.println("[BrokerServer] connection 连接关闭 ,客户端地址: " + clientSocket.getInetAddress().toString()
                        + " : " + clientSocket.getPort());
            } catch (ClassNotFoundException e) {
                throw new RuntimeException(e);
            } catch (MqException e) {
                throw new RuntimeException(e);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            //关闭资源
            try {
                //当前连接处理完之后,需要关闭Socket
                clientSocket.close();
                //把当前socket对应的所有channel也删除了
                clearCloseSessions(clientSocket);
            } catch (IOException e) {
                e.printStackTrace();
            }
            //删除sessions中客户端和服务器建立的连接
        }
    }

    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        byte[] payload = new byte[request.getLength()];
        int n = dataInputStream.read(payload);
        if (n != request.getLength()) {
            throw new IOException("读取请求格式出错!");
        }
        request.setPayload(payload);
        return request;
    }

    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        // 这个刷新缓冲区也是重要的操作!!
        dataOutputStream.flush();
    }

    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        // 1. 把 request 中的 payload 做一个初步的解析.
        BasicArgs BasicArgs = (BasicArgs) BinaryTool.fromBytes(request.getPayload());
        System.out.println("[Request] rid=" + BasicArgs.getRid() + ", channelId=" + BasicArgs.getChannelId()
                + ", type=" + request.getType() + ", length=" + request.getLength());
        // 2. 根据 type 的值, 来进一步区分接下来这次请求要干啥.
        boolean ok = true;
        if (request.getType() == 0x1) {
            // 创建 channel
            sessions.put(BasicArgs.getChannelId(), clientSocket);
            System.out.println("[BrokerServer] 创建 channel 完成! channelId=" + BasicArgs.getChannelId());
        } else if (request.getType() == 0x2) {
            // 销毁 channel
            sessions.remove(BasicArgs.getChannelId());
            System.out.println("[BrokerServer] 销毁 channel 完成! channelId=" + BasicArgs.getChannelId());
        } else if (request.getType() == 0x3) {
            // 创建交换机. 此时 payload 就是 ExchangeDeclareArgs 对象了.
            ExchangeDeclareArgs Args = (ExchangeDeclareArgs) BasicArgs;
            ok = virtualHost.exchangeDeclare(Args.getExchangeName(), Args.getType(),
                    Args.isDurable(), Args.isAutoDelete(), Args.getArgs());
        } else if (request.getType() == 0x4) {
            ExchangeDeleteArgs Args = (ExchangeDeleteArgs) BasicArgs;
            ok = virtualHost.exchangeDelete(Args.getExchangeName());
        } else if (request.getType() == 0x5) {
            QueueDeclareArgs Args = (QueueDeclareArgs) BasicArgs;
            ok = virtualHost.queueDeclare(Args.getQueueName(), Args.isDurable(), Args.isAutoDelete(), Args.getArgs());
        } else if (request.getType() == 0x6) {
            QueueDeleteArgs Args = (QueueDeleteArgs) BasicArgs;
            ok = virtualHost.queueDelete((Args.getQueueName()));
        } else if (request.getType() == 0x7) {
            BindingDeclareArgs Args = (BindingDeclareArgs) BasicArgs;
            ok = virtualHost.bindingDeclare(Args.getExchangeName(), Args.getQueueName(), Args.getBindingKey());
        } else if (request.getType() == 0x8) {
            BindingDeleteArgs Args = (BindingDeleteArgs) BasicArgs;
            ok = virtualHost.bindingDelete( Args.getExchangeName(),Args.getQueueName());
        } else if (request.getType() == 0x9) {
            BasicPublishArgs Args = (BasicPublishArgs) BasicArgs;
            ok = virtualHost.basicPublish(Args.getExchangeName(), Args.getRoutingKey(),
                    Args.getBasicProperties(), Args.getBody());
        } else if (request.getType() == 0xa) {
            BasicConsumerArgs Args = (BasicConsumerArgs) BasicArgs;
            ok = virtualHost.basicConsume(Args.getConsumerTag(), Args.getQueueName(), Args.isAutoAck(),
                    new Consumer() {
                        //这个回调函数要做的工作, 就是把服务器收到的消息可以直接推送回对应的消费者客户端
                        //此处 consumerTag 其实是 channelId. 根据 channelId 去 sessions 中查询,
                        //  就可以得到对应的socket 对象了, 从而可以往里面发送数据了
                        @Override
                        public void handlerDeliver(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                            // 先知道当前这个收到的消息, 要发给哪个客户端.
                            // 此处 consumerTag 其实是 channelId. 根据 channelId 去 sessions 中查询, 就可以得到对应的
                            // socket 对象了, 从而可以往里面发送数据了
                            // 1. 根据 channelId 找到 socket 对象
                            Socket clientSocket = sessions.get(consumerTag);
                            if (clientSocket == null || clientSocket.isClosed()) {
                                throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭!");
                            }
                            // 2. 构造响应数据
                            SubScribeReturns subScribeReturns = new SubScribeReturns();
                            subScribeReturns.setChannelId(consumerTag);
                            subScribeReturns.setRid(""); // 由于这里只有响应, 没有请求, 不需要去对应. rid 暂时不需要.
                            subScribeReturns.setOk(true);
                            subScribeReturns.setConsumerTag(consumerTag);
                            subScribeReturns.setBasicProperties(basicProperties);
                            subScribeReturns.setBody(body);
                            byte[] payload = BinaryTool.toByte(subScribeReturns);
                            Response response = new Response();
                            // 0xc 表示服务器给消费者客户端推送的消息数据.
                            response.setType(0xc);
                            // response 的 payload 就是一个 SubScribeReturns
                            response.setLength(payload.length);
                            response.setPayload(payload);
                            // 3. 把数据写回给客户端.
                            //    注意! 此处的 dataOutputStream 这个对象不能 close !!!
                            //    如果 把 dataOutputStream 关闭, 就会直接把 clientSocket 里的 outputStream 也关了.
                            //    此时就无法继续往 socket 中写入后续数据了.
                            DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                            writeResponse(dataOutputStream, response);
                        }
                    });
        } else if (request.getType() == 0xb) {
            // 调用 basicAck 确认消息.
            BasicAckArgs Args = (BasicAckArgs) BasicArgs;
            ok = virtualHost.basicAck(Args.getQueueName(), Args.getMessageId());
        } else {
            // 当前的 type 是非法的.
            throw new MqException("[BrokerServer] 未知的 type! type=" + request.getType());
        }
        // 3. 构造响应
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setChannelId(BasicArgs.getChannelId());
        basicReturns.setRid(BasicArgs.getRid());
        basicReturns.setOk(ok);
        byte[] payload = BinaryTool.toByte(basicReturns);
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[Response] rid=" + basicReturns.getRid() + ", channelId=" + basicReturns.getChannelId()
                + ", type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

    private void clearCloseSessions(Socket clientSocket) {
        // 这里要做的事情, 主要就是遍历上述 sessions hash 表, 把该被关闭的 socket 对应的键值对, 统统删掉.
        List<String> toDeleteChannelId = new ArrayList<>();
        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) {
                // 不能在这里直接删除!!!
                // 这属于使用集合类的一个大忌!!! 一边遍历, 一边删除!!!
                // sessions.remove(entry.getKey());
                toDeleteChannelId.add(entry.getKey());
            }
        }
        for (String channelId : toDeleteChannelId) {
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteChannelId);
    }
}
